# -*- coding: utf-8 -*-
"""
项目名称：NoSql_kafka_project
文件名称：kafka_producer.py
创建时间：2025-06-05 16:21:23

系统用户：Administrator
作　　者：無以菱
联系邮箱：huangjing2001.guet@qq.com
功能描述：
- 功能：模拟生成支付订单数据，包含订单ID、支付金额、支付方式和创建时间
- 配置：连接到Kafka集群（192.168.10.10:9092等）
- 主题：real_time_orders
"""
from kafka import KafkaProducer  
import json  
import time  
import random  

# 初始化Kafka生产者实例
# bootstrap_servers: 指定Kafka集群地址，使用多个服务器实现高可用
# value_serializer: 定义如何序列化消息值，这里将Python对象转换为JSON字符串并编码为UTF-8
producer = KafkaProducer(
    bootstrap_servers=['192.168.10.10:9092','192.168.10.11:9092','192.168.10.12:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)
# 定义消息主题名称，所有订单数据将发送到此主题
topic = "real_time_orders"

def generate_order():
    """
    生成一个模拟的订单数据
    
    返回:
        dict: 包含订单ID、支付金额、支付方式和创建时间的字典
    """
    return {
        "ORDER_ID": random.randint(1000, 9999),  # 随机生成4位数的订单ID
        "PAY_AMOUNT": round(random.uniform(10, 1000), 2),  # 随机生成10到1000之间的支付金额，保留2位小数
        "PAYWAY": random.choice(["支付宝", "微信", "银行卡"]),  # 随机选择一种支付方式
        "CREATE_TIME": int(time.time())  # 使用当前时间戳作为订单创建时间
    }

def run_kafka_producer():
    """
    运行Kafka生产者，持续生成订单数据并发送到Kafka
    
    该函数会进入一个无限循环，每秒生成一条新的订单数据并发送到Kafka主题
    """
    print("Kafka生产者启动，每秒生成1条数据...")
    while True:  # 无限循环，持续生成数据
        order = generate_order()  # 生成一个新的订单
        producer.send(topic, value=order)  # 将订单数据发送到Kafka主题
        print(f"已生成订单数据: {order}")  # 打印生成的订单数据
        time.sleep(1)  # 暂停1秒，控制数据生成频率

if __name__ == "__main__":
    # 单独运行此文件时执行
    print("正在启动Kafka生产者进行调试...")
    try:
        run_kafka_producer()  # 启动Kafka生产者
    except KeyboardInterrupt:
        print("Kafka生产者已停止")  # 捕获键盘中断信号（Ctrl+C），优雅地退出
    except Exception as e:
        print(f"Kafka生产者运行出错: {e}")  # 捕获并打印其他异常
